Skip to content

Add streaming API #463

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Merged
merged 23 commits into from
Dec 9, 2024
Merged

Add streaming API #463

merged 23 commits into from
Dec 9, 2024

Conversation

@jpsamaroo
Copy link
Member

@jpsamaroo jpsamaroo commented Dec 21, 2023 •
edited
Loading

Adds a spawn_streaming task queue to transform tasks into continuously-executing equivalents that automatically take from inputs streams/channels and put their result to an output stream/channel. Useful for processing tons of individual elements of some large (or infinite) collection.

Todo:

  • Migrate streams on first use
  • Add per-task input buffering to Stream object
  • Add no-allocation ring buffer for process-local put/take to Stream
  • Make buffering amount configurable
  • Add API for constructing streams based on inferred return type, desired buffer size, and source/destination
  • Allow finish_stream(xyz; return=abc) to return custom value (else nothing)
  • Upstream MemPool migration changes (Add DRef migration support JuliaData/MemPool.jl#80)
  • Add docs
  • Add tests
  • (Optional) Adapt ring buffer to support server-local put/take (use mmap?)
  • (Optional) Make value fetching configurable
  • (Optional) Support a waitany-style input stream, taking inputs from multiple tasks
  • (Optional) take! from input streams concurrently, and waitall on them before continuing
  • (Optional) put! into output streams concurrently, and waitall on them before continuing
  • (Optional) Allow using default or previously-cached value if sender not ready
  • (Optional) Allow dropping stream values (after timeout, receiver not ready, over-pressured, etc.)
  • (Optional) Add utility for tracking stream transfer rates (Add streaming throughput monitor #494)
  • (Optional) Add programmable behavior on upstream/downstream Stream closure (how should errors/finishing propagate?)

@jpsamaroo jpsamaroo marked this pull request as draft December 21, 2023 15:33
@JamesWrigley JamesWrigley mentioned this pull request Mar 13, 2024
Merged
@JamesWrigley JamesWrigley mentioned this pull request Mar 20, 2024
Merged
@JamesWrigley
Copy link
Collaborator

JamesWrigley commented Mar 29, 2024

Am I correct in thinking that all the necessary items except for tests are complete?

@jpsamaroo
Copy link
Member Author

jpsamaroo commented Mar 31, 2024

Generally yes, I think we're pretty close to this being merge-ready. There are some remaining TODOs that I need to finish, but most are reasonably small. I could definitely use help with writing tests - just validating that we can run various kinds of pipelines and that they work across multiple workers would be really useful.

@JamesWrigley JamesWrigley mentioned this pull request Apr 1, 2024
Merged
@JamesWrigley JamesWrigley force-pushed the jps/stream2 branch 2 times, most recently from ed89a7f to 3274093 Compare August 3, 2024 16:57
@JamesWrigley JamesWrigley mentioned this pull request Aug 16, 2024
Merged
@jpsamaroo jpsamaroo marked this pull request as ready for review December 3, 2024 22:41
@jpsamaroo jpsamaroo mentioned this pull request Dec 3, 2024
Closed
@jpsamaroo jpsamaroo force-pushed the jps/stream2 branch 2 times, most recently from d259f57 to 59a0371 Compare December 4, 2024 17:49
JamesWrigley and others added 19 commits December 9, 2024 05:33
@JamesWrigley @jpsamaroo
Delete Dagger.cleanup()
b5fbb0f 
Because it doesn't actually do anything now.
@JamesWrigley @jpsamaroo
Use procs() when initializing EAGER_CONTEXT
080d7bc 
Using `myid()` with `workers()` meant that when the context was initialized with
a single worker the processor list would be: `[OSProc(1), OSProc(1)]`. `procs()`
will always include PID 1 and any other workers, which is what we want.
@jpsamaroo
Add metadata to EagerThunk
ebdfda4
@jpsamaroo
Sch: Allow occupancy key to be Any
fcff991
@jpsamaroo
Add a --verbose option to runtests.jl
5044676
@jpsamaroo
task-tls: Refactor into DTaskTLS struct
ed2493c
@jpsamaroo
cancellation: Add cancel token support
c07db4c
@jpsamaroo
task-tls: Tweaks and fixes, task_id helper
2af3f10
@jpsamaroo
cancellation: Add graceful vs. forced
6da10fa
@jpsamaroo
cancellation: Wrap InterruptException in DTaskFailedException
2515c36
@jpsamaroo
Sch: Add unwrap_nested_exception for DTaskFailedException
2b70bf3
@jpsamaroo
Add task_id for DTask
3638866
@jpsamaroo
dagdebug: Always yield to avoid heisenbugs
2d75603
@jpsamaroo
tests: Add offline mode
1c806e9
@jpsamaroo
dagdebug: Add JULIA_DAGGER_DEBUG config variable
5233385
@jpsamaroo
options: Add internal helper to strip all options
2d20762
@jpsamaroo
tests: Test DTaskFailedException inner type
32b2ccd
@jpsamaroo
Sch: Skip not-yet-inited workers
a5b663e
@JamesWrigley @jpsamaroo
Bump MemPool compat
577e179
@jpsamaroo jpsamaroo mentioned this pull request Dec 9, 2024
Closed
@jpsamaroo jpsamaroo merged commit 62f8307 into master Dec 9, 2024
8 of 11 checks passed
@jpsamaroo jpsamaroo deleted the jps/stream2 branch December 9, 2024 21:40
@jpsamaroo
Copy link
Member Author

jpsamaroo commented Dec 9, 2024

Thanks so much to @JamesWrigley and @davidizzle for making this a reality! ❤️

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

data movement eager api enhancement needs docs needs tests performance streaming

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@jpsamaroo @JamesWrigley